package com.iot.gateway.factory;

import com.iot.gateway.dto.MqttClientParam;
import com.iot.gateway.exception.CommonException;
import java.util.concurrent.ConcurrentHashMap;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * @Description
 * @Author xinggq
 * @Date 2025/5/27
 */
@Slf4j
public class MqttFactory {

  private final ConcurrentHashMap<String, MqttClient> mqClientCache = new ConcurrentHashMap<>();


  /**
   * 获取mqtt客户端，如果客户端不存在，则初始化客户端
   * @param param
   * @return
   */
  public MqttClient getOrInitClient(MqttClientParam param) {
    MqttClient client = mqClientCache.get(param.getClientId());
    if (client == null) {
      // 执行初始化流程
      client = initClient(param);
      mqClientCache.put(param.getClientId(), client);
      return client;
    }
    if (client.isConnected()) {
      return client;
    }
    try {
      client.reconnect();
      return client;
    } catch (MqttException e) {
      log.error("mqtt客户端:{}重连失败", param.getClientId(), e);
      throw new CommonException(String.format("mqtt客户端:%s重连失败", param.getClientId()));
    }
  }

  /**
   * 移除并释放mqtt客户端
   * @param clientId
   * @return
   */
  public boolean removeAndReleaseClient(String clientId) {
    MqttClient client = mqClientCache.get(clientId);
    if (client == null) {
      log.info("mqtt客户端:{}不存在", clientId);
      return true;
    }
    try {
      if (client.isConnected()) {
        client.disconnect(); // 显式断开连接
      }
      mqClientCache.remove(clientId);
      return true;
    } catch (MqttException e) {
      log.error("MQTT客户端:{}断开连接失败", clientId, e);
      throw new CommonException(String.format("mqtt客户端:%s断开连接失败", clientId));
    }
  }

  /**
   * 初始化mqtt客户端
   * @param param
   * @return
   */
  private MqttClient initClient(MqttClientParam param) {

    try {
      MqttClient client = new MqttClient(param.getServerUrl(), param.getClientId(),
          new MemoryPersistence());
      MqttConnectOptions options = new MqttConnectOptions();
      // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，
      // 这里设置为true表示每次连接到服务器都以新的身份连接
      options.setCleanSession(true);
      options.setUserName(param.getUsername());
      options.setPassword(param.getPassword().toCharArray());
      options.setConnectionTimeout(param.getConnectionTimeout());
      options.setKeepAliveInterval(param.getKeepAliveInterval());
      //设置断开后重新连接
      options.setAutomaticReconnect(true);
      client.connect(options);
      // 订阅
      client.subscribe(param.getSubscribeTopic(), param.getListener());
      return client;
    } catch (MqttException e) {
      log.error("mqtt客户端:{}初始化失败,", param.getClientId(), e);
      throw new CommonException(String.format("mqtt客户端:%s初始化失败", param.getClientId()));
    }
  }


}
